/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package middleware;

import flight.client.FlightServiceProxy;
import flight.server.FlightClientProxy;
import flight.service.FlightService;
import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.SocketException;
import java.net.SocketTimeoutException;
import java.net.UnknownHostException;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * The main class of middleware layer. For sending and receiving message
 * @author HO HOANG HUNG
 */
public class Naming {
    //Data buffer size of datagram packet
    private static final int DATALENGTH = 4096;
    //Default listening port
    private static final int LISTENPORT = 2222;
    //True if at-most-once invocation semantics
    private boolean atmost1 = false;
    //Datagram socket for sending/receiving message
    private DatagramSocket socket = null;
    //For upper layer
    private Remote upperobject;
    //For simulation of message lost
    private float threshold = 1;
    //Time out in second
    private int timeout = 5;
    //Listening to request
    private NamingThread nmthread = null;
    //Histories for duplicate checking & stored results
    private HashMap<InetAddress, Integer> processed = null;
    private HashMap<InetAddress, Object> processedresult = null;
    //Message Id to be used at client
    private byte messageid = 0;
    private int port = LISTENPORT;
    // to start/stop the listen-for-request thread
    private boolean listen = true;

    /**
     * Default constructor
     */
    public Naming() {
        nmthread = new NamingThread();
    }

    public Naming(boolean am1, float threshold) {
        // port = p;
        atmost1 = am1;
        if ((0 < threshold) && (threshold <= 1)) {
            this.threshold = threshold;
        }
        nmthread = new NamingThread();
    }

    public float getThreshold() {
        return threshold;
    }

    /**
     * For simulation of lost request/reply message, lower value means more lost message
     * @param th value from 0.0f to 1.0f
     */
    public void setThreshold(float th) {
        if ((0 < th) && (th <= 1)) {
            threshold = th;
        }
    }

    public int getTimeout() {
        return timeout;
    }

    public int getPort() {
        return port;
    }

    /**
     * (step 3)
     * Set port if need to. This is the port for listening for request.
     * @param p
     */
    public void setPort(int p) {
        if (p > 0) {
            port = p;
        }
    }

    /**
     * Set Timeout. Unit is second
     * @param to
     */
    public void setTimeout(int to) {
        if (to > 0) {
            //   System.out.println("TIMEOUT = "+to);
            timeout = to;
        }
    }

    public boolean isAtMostOnce() {
        return atmost1;
    }

    public boolean isAtLeastOnce() {
        return !atmost1;
    }

    /**
     * At most once or At-least-once semantic
     * @param am
     */
    public void setAtMostOnce(boolean am) {
        atmost1 = am;
    }

    /**
     * (step 1)
     * Set the object from above layer that provide service.
     * @param newo object that provides service
     */
    public void setUpperObject(Remote newo) {
        upperobject = newo;
    }

    /**
     * (step 4) start listening for request. Object that provides service is register here if haven't
     * been registered
     * @param object
     */
    public void createRegistry(Remote object) {
        upperobject = object;
        createRegistry();
    }

    /**
     * step 4: start listening for request.
     */
    public void createRegistry() {
        listen = true;
        processed = new HashMap<InetAddress, Integer>(2);
        processedresult = new HashMap<InetAddress, Object>(2);
        nmthread = new NamingThread();
        nmthread.start();
    }

    /**
     * To use when client wants to send request to server
     * String parameter should not exceed 127 characters in length.
     * @param iptext
     * @param methodName should not exceed 127 characters in length.
     * @param paras
     * @return null if the method has no return value
     */
    public Object send(String iptext, String methodName, Object[] paras) {
        //Convert text IP address to raw
        int numofsend = 0;
        String[] iptexta = iptext.split("\\.");
        byte[] ipbyte = new byte[iptexta.length];
        for (int i = 0; i < ipbyte.length; i++) {
            int tempbyte = Integer.parseInt(iptexta[i]);
            ipbyte[i] = (byte) tempbyte;
        }
        InetAddress recvip = null;
        try {
            recvip = InetAddress.getByAddress(ipbyte);
        } catch (UnknownHostException ex) {
            Logger.getLogger(Naming.class.getName()).log(Level.SEVERE, null, ex);
        }
        //Prepare send socket
        DatagramSocket ssocket = null;
        try {
            ssocket = new DatagramSocket();
            ssocket.setSoTimeout(timeout * 1000);
        } catch (SocketException ex) {
            Logger.getLogger(Naming.class.getName()).log(Level.SEVERE, null, ex);
        }

        //Prepare data
        byte[] buf = new byte[DATALENGTH];
        messageid = (byte) ((messageid + 1) % 128);
        int offset = 0;
        buf[offset++] = messageid;
        byte[] methodnamebyte = methodName.getBytes();
        buf[offset++] = (byte) (methodnamebyte.length);
        for (int i = 0; i < methodnamebyte.length; i++) {
            buf[offset++] = methodnamebyte[i];
        }
        try {
            offset = Serialization.serializeParameters(paras, buf, offset);
        } catch (Exception ex) {
            ex.printStackTrace();
        }
        //There should be a limit for the number of tries for reply
        int received = 0;
        int roffset = 0;
        byte[] recvbuf;
        DatagramPacket recvpacket = null;
        while (received < 10) {
            numofsend++;
            try {
                //Send the message
                if (canSend()) {
                    DatagramPacket packet = new DatagramPacket(buf, buf.length, recvip, port);
                    ssocket.send(packet);
                }
            } catch (IOException ex) {
                Logger.getLogger(Naming.class.getName()).log(Level.SEVERE, null, ex);
            }
            // get response
            recvbuf = new byte[DATALENGTH];
            recvpacket = new DatagramPacket(recvbuf, recvbuf.length);
            try {
                ssocket.receive(recvpacket);
                //Set to a number > 10 to indicate that we already received instead
                // of timeout because of multiple tries
                received = 20;
            } catch (SocketTimeoutException ex) {
                //At most once at least 1 timeout here.
                System.out.println("Timeout");
                received++;
            } catch (IOException ex) {
                Logger.getLogger(Naming.class.getName()).log(Level.SEVERE, null, ex);
            }
        }
        Object result = null;
        //Actually received the reply
        if (received == 20) {
            System.out.println("Received reply after sending " + numofsend + " request(s)");
            recvbuf = recvpacket.getData();
            int mssgid = recvbuf[roffset++];

            ssocket.close();
            try {
                result = Serialization.deserializeObject(recvbuf, new MutableInteger(roffset));
            } catch (Exception ex) {
                Logger.getLogger(Naming.class.getName()).log(Level.SEVERE, null, ex);
            }
            //Get out of the loop because of timeout
        }else{

            System.out.println("Cannot received reply after multiple tries");
        }
        return result;
    }

    /**
     * Retrieve the proxy object
     * @param ip
     * @param remoteName
     * @return
     */
    public Remote lookup(String ip, String remoteName) {
        if (remoteName.equalsIgnoreCase("FlightService")) {
            FlightService ser = new FlightServiceProxy(ip, this);
            return ser;
        } else if (remoteName.equalsIgnoreCase("FlightClient")) {
            Callback cb = new FlightClientProxy(ip, this);
            return cb;
        } else {
            return null;
        }
    }

    /**
     * Stop listening for request on the default port
     */
    public void releaseRegistry() {
        System.out.println("Release the registry...");
        listen = false;
    }

    /**
     * For simulation of lost request/reply message
     * @return false if message will be 'lost', true otherwise
     */
    private boolean canSend() {
        if (threshold == 1) {
            return true;
        } else {
            float rand = (float) Math.random();
            if (rand > threshold) {
                System.out.println("The send message will be lost");
                return false;
            } else {
                return true;
            }
        }
    }

    /**
     * We will listen for request on default port in this thread
     */
    private class NamingThread extends Thread {

        @Override
        public void run() {
            try {
                //Initialize the socket if it is null or closed
                if ((socket == null)||(socket.isClosed())) {
                    System.out.println("Starting listening for request at port "+port);
                    socket = new DatagramSocket(port);
                }
                //Keep listening until the registry is released
                while (listen) {
                    byte[] buf = new byte[DATALENGTH];
                    // receive request
                    DatagramPacket packet = new DatagramPacket(buf, buf.length);
                    socket.receive(packet);
                    buf = packet.getData();
                    InetAddress from = packet.getAddress();
                    int rbufpointer = 0;
                    byte mssgid = buf[rbufpointer++];
                    Object result = null;
                    //check for duplicate
                    if (atmost1 && inHistory(from, mssgid)) {
                        System.out.println("This request has already been processed. Retrieve result from history");
                        result = processedresult.get(from);
                    } else {
                        // figure out response
                        //Construct the method to invoke
                        int methodnamel = buf[rbufpointer++];
                        byte[] mthnamearray = new byte[methodnamel];
                        for (int i = 0; i < mthnamearray.length; i++) {
                            mthnamearray[i] = buf[rbufpointer++];
                        }
                        String methodname = new String(mthnamearray);
                        Object[] paras = Serialization.deserializeParameters(buf, rbufpointer);
                        //if it's a register callback method, add a callback object to it
                        if (methodname.toLowerCase().indexOf("registercallback") >= 0) {
                            System.out.println("Naming Thread: register callback from " + from.getHostAddress());
                            if (methodname.toLowerCase().indexOf("flightclient") >= 0) {
                                paras[2] = lookup(from.getHostAddress(), "FlightClient");
                            }
                        }

                        result = upperobject.invoke(methodname, paras);
                        //store to history if in at-most-once semantics
                        if (atmost1) {
                            System.out.println("At-most-once: store result to histories");
                            processed.put(from, Integer.valueOf(mssgid));
                            processedresult.put(from, result);
                        }
                    }
                    // send the response to the client at "address" and "port"
                    int port = packet.getPort();
                    int offset = 0;
                    byte[] sbuf = new byte[DATALENGTH];
                    sbuf[offset++] = mssgid;
                    Serialization.serializeObject(result, sbuf, offset);
                    if (canSend()) {
                        DatagramPacket spacket = new DatagramPacket(sbuf, sbuf.length, from, port);
                        socket.send(spacket);
                    }
                }
                System.out.println("Close the socket ...");
                socket.close();
            } catch (Exception ex) {
                ex.printStackTrace();
            }
        }

        /**
         * Check whether the message with this message identifier (IP+id) has already
         * been processed.
         * @param from client IP address
         * @param mssgid local message id (kept track by each client)
         * @return true if already been processed, false otherwise.
         */
        private boolean inHistory(InetAddress from, byte mssgid) {
            Integer mid = processed.get(from);
            if (mid == null) {//Never communicate with this address before
                return false;
            } else {
                if (mid == mssgid) {
                    return true;
                } else {
                    return false;
                }
            }
        }
    }
}
